Multi-Agent Architectures

import asyncio
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from customer_service import customer_service_node
from recommend import recommend_node
from reimburse import reimburse_node
from state import State
from supervisor import supervisor_node

builder = StateGraph(State)
builder.add_node("supervisor_node", supervisor_node)
builder.add_node("recommend_node", recommend_node)
builder.add_node("customer_service_node", customer_service_node)
builder.add_node("reimburse_node", reimburse_node)

# ===== 流程控制 =====
def estimate(state: State) -> str | None:
    if state.phase == "dispatch":  # 任务分发阶段
        if state.type == "recommend":
            return "recommend_node"
        elif state.type == "customer_service":
            return "customer_service_node"
        elif state.type == "reimburse":
            return "reimburse_node"
        return None
    else:  # 结果汇总阶段
        return "END"  # 汇总完成 → END

# START → supervisor(dispatch)
builder.add_edge(START, "supervisor_node")

# supervisor(dispatch) → Agent / supervisor(gather) → END
builder.add_conditional_edges("supervisor_node", estimate, {
    "recommend_node": "recommend_node",
    "customer_service_node": "customer_service_node",
    "reimburse_node": "reimburse_node",
    "END": END
})

# Agent → supervisor(gather)
builder.add_edge("recommend_node", "supervisor_node")
builder.add_edge("customer_service_node", "supervisor_node")
builder.add_edge("reimburse_node", "supervisor_node")

# ===== 编译 & 测试 =====
graph = builder.compile()
graph.get_graph().draw_png('./graph.png')

# 使用异步方式调用
async def main():
    # content1 = await graph.ainvoke({"messages": ["我今天中午该吃什么啊"]})
    # content1["messages"][-1].pretty_print()

    # content2 = await graph.ainvoke({"messages": ["我点的外卖不想要了，我要退钱"]})
    # content2["messages"][-1].pretty_print()
    content3 = await graph.ainvoke({"messages": ["我要报销这张餐补发票"]})
    content3["messages"][-1].pretty_print()

# 运行异步主函数
asyncio.run(main())

import asyncio
import dotenv
from loguru import logger
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain import hub
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import AIMessage
from langchain.agents import create_react_agent, create_openai_tools_agent, AgentExecutor
from llm import llm
from state import State
async def recommend_node(state: State) -> State:
    dotenv.load_dotenv()
    # 1️⃣ 加载服务器配置
    mcp_client = MultiServerMCPClient({
        "howtocook-mcp": {
            "transport": "sse",
            "url": "https://dashscope.aliyuncs.com/api/v1/mcps/how-to-cook/sse",
            "headers": {
                "Authorization": "Bearer " + dotenv.get_key(".env", "BAILIAN_API_KEY"),
            }
        }
    })
    # 2️⃣ 初始化 MCP 客户端并获取工具
    tools = await mcp_client.get_tools()
    logger.info(f"✅ 已加载 {len(tools)} 个 MCP 工具： {[t.name for t in tools]}")
    # 3️⃣ 初始化语言模型、提示模板和代理执行器
    prompt = hub.pull("hwchase17/openai-tools-agent")
    agent = create_openai_tools_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    logger.info("执行美食推荐agent")
    result = await agent_executor.ainvoke({"input": state.messages})
    # logger.info(f"美食推荐结果: {result}")
    return State(messages=state.messages + [AIMessage(f"{result['output']}")],
                 type="recommend", phase="gather")

from typing import TypedDict, Annotated, Optional
from langgraph.constants import START, END
from langgraph.graph import add_messages, StateGraph
from loguru import logger
from pydantic import BaseModel, Field
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from llm import chat_model
from state import State

def supervisor_node(state: State) -> State | None:
    logger.info(f"[supervisor_node] 当前阶段: {state.phase}, State: {state}")

    if state.phase == "dispatch":
        # 分发阶段 -> 分类问题
        chat_prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的客服助手，专门负责对用户提出的问题进行分类，并将任务分发给其他Agent执行。
                            如果用户的问题和食谱推荐有关，那就返回recommend。
                            如果用户的问题和外卖问题有关，那就返回customer_service。
                            如果用户的问题和报销发票有关，那就返回reimburse_extract。
                            除了上述选项外，不要返回其他的内容。"""),
            ("human", "用户提出的问题是：{question}")
        ])
        parser = StrOutputParser()
        chain = chat_prompt | chat_model | parser
        task_type = chain.invoke({"question": state.messages})
        logger.info(f"问题分类结果: {task_type}")
        return State(messages=state.messages, type=task_type.strip(), phase="dispatch")
    elif state.phase == "gather":
        # 汇总阶段 -> 整理子Agent结果
        return State(messages=state.messages, type="summary", phase="done")
    else:
        return None